package sinktest

import Source.SourceTest.SensorReading
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer


// flink写入kafka
object KafkaSinkTest {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // "E:\\Flinklearn\\atguigu\\src\\main\\resources\\sensor.txt"
    val inputPath: String = "file:///usr/local/src/flink/sensor.txt"
    val inputDataStream: DataStream[String] = env.readTextFile(inputPath)

    // 转换成样例类类型
    val ds1 = inputDataStream.map(
      data => {
        val fields: Array[String] = data.split(",")
        SensorReading(fields(0), fields(1).toLong, fields(2).toDouble).toString
      })

    ds1.print()

    ds1.addSink( new FlinkKafkaProducer[String]("master:9092","sinktest",new SimpleStringSchema()))


    env.execute("kafka sink test")
  }

}
